package org.apache.rocketmq.remoting.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;

import java.net.InetSocketAddress;
import java.time.Duration;
import java.util.concurrent.*;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/25
 * @方法描述：Netty构建的Namesrv的服务端
 */
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {


    private static final Logger TRAFFIC_LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_TRAFFIC_NAME);

    private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);


    //下面这些都是构建Netty服务器要用到的成员变量，大家应该很熟悉了吧？
    //Netty服务器的启动器
    private final ServerBootstrap serverBootstrap;

    //处理SocketChannel的IO事件的循环组
    private final EventLoopGroup eventLoopGroupSelector;
    //处理连接事件的循环组
    private final EventLoopGroup eventLoopGroupBoss;
    //Netty服务器要用到的配置信息
    private final NettyServerConfig nettyServerConfig;
    //这个执行器会被用来执行一些公共请求事件
    private final ExecutorService publicExecutor;
    //这个定时任务执行器会定期打印请求Namesrv服务端接收到的请求和返回响应的次数，便于开发者对程序执行性能进行分析
    private final ScheduledExecutorService scheduledExecutorService;
    //这个监听器就会被定义在NamesrvController中的brokerHousekeepingService成员变量赋值，在当前类的NettyConnectManageHandler内部类中
    //这个channelEventListener监听器就会被用到
    private final ChannelEventListener channelEventListener;

    //Netty的时间轮组件
    private final HashedWheelTimer timer = new HashedWheelTimer(r -> new Thread(r, "ServerHouseKeepingService"));

    //这个执行器在第一版本代码中，这个执行器组就是用来执行设置到channel中的handler操作的
    private DefaultEventExecutorGroup defaultEventExecutorGroup;

    //这个map存储的是端口号和对应的Netty服务器的映射，在第一版本代码中还不会出现这种情况，等后面我会为大家讲解这种情况出现的原因
    private final ConcurrentMap<Integer/*Port*/, NettyRemotingAbstract> remotingServerTable = new ConcurrentHashMap<>();

    //Netty服务器的编码器
    private NettyEncoder encoder;
    //Netty服务器接收到的channel的管理器
    private NettyConnectManageHandler connectionManageHandler;

    //Netty服务端的channel处理器，这个处理器就是专门处理接收到的网络消息的
    private NettyServerHandler serverHandler;
    //这个处理器会记录Namesrv接收不同请求和回复响应的次数
    private RemotingCodeDistributionHandler distributionHandler;


    //构造方法
    public NettyRemotingServer(final NettyServerConfig nettyServerConfig) {
        this(nettyServerConfig, null);
    }



    //构造方法
    public NettyRemotingServer(final NettyServerConfig nettyServerConfig, final ChannelEventListener channelEventListener) {
        //这里有一个比较重要的操作，那就是从Netty服务端的配置类中得到了单向信号量和异步信号量的次数，然后把它们传递给父类使用了
        //这两个信号量在第一版本代码中还用不到，在第二版本代码中就会用到了，它们主要是用来限制同一时刻Namesrv服务器对外发送单项请求和异步请求的数量
        //防止请求过多挤占服务器资源
        super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
        this.serverBootstrap = new ServerBootstrap();
        this.nettyServerConfig = nettyServerConfig;
        this.channelEventListener = channelEventListener;
        this.publicExecutor = buildPublicExecutor(nettyServerConfig);
        this.scheduledExecutorService = buildScheduleExecutor();
        this.eventLoopGroupBoss = buildBossEventLoopGroup();
        this.eventLoopGroupSelector = buildEventLoopGroupSelector();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：创建channel的IO事件循环组
     */
    private EventLoopGroup buildEventLoopGroupSelector() {
        //判断是否启用Epoll模式
        if (useEpoll()) {
            return new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactoryImpl("NettyServerEPOLLSelector_"));
        } else {//不启用的话则创建NIO的事件循环组，事件循环组使用的线程数量也要从配置信息类中获得
            return new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactoryImpl("NettyServerNIOSelector_"));
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：创建处理连接事件的循环组
     */
    private EventLoopGroup buildBossEventLoopGroup() {
        if (useEpoll()) {
            return new EpollEventLoopGroup(1, new ThreadFactoryImpl("NettyEPOLLBoss_"));
        } else {
            return new NioEventLoopGroup(1, new ThreadFactoryImpl("NettyNIOBoss_"));
        }
    }


    //判断是否使用Epoll模式的方法
    private boolean useEpoll() {
        //先判断是否为linux平台，再判断是否开启了Epoll模式
        return NetworkUtil.isLinuxPlatform()
                && nettyServerConfig.isUseEpollNativeSelector()
                && Epoll.isAvailable();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：创建公共请求执事件行器的方法
     */
    private ExecutorService buildPublicExecutor(NettyServerConfig nettyServerConfig) {
        //从Netty配置信息类中获得处理公共请求执事件执行器线程数量
        int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
        //如果为0，就意味着用户没有配置这个线程数量
        if (publicThreadNums <= 0) {
            //那就使用默认值4
            publicThreadNums = 4;
        }
        //创建执行器
        return Executors.newFixedThreadPool(publicThreadNums, new ThreadFactoryImpl("NettyServerPublicExecutor_"));
    }

    //创建定时任务调度器的方法
    private ScheduledExecutorService buildScheduleExecutor() {
        //这个定时任务调度器是单线程的
        return ThreadUtils.newScheduledThreadPool(1,
                new ThreadFactoryImpl("NettyServerScheduler_", true),
                new ThreadPoolExecutor.DiscardOldestPolicy());
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：启动Netty服务器的方法
     */
    @Override
    public void start() {
        //在这里创建了默认的执行器组，从nettyServerConfig配置信息对象中得到了执行器组要使用的线程数
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(nettyServerConfig.getServerWorkerThreads(),
                new ThreadFactoryImpl("NettyServerCodecThread_"));

        //为服务端接收到的客户端channel准备可共享的handler处理器
        prepareSharableHandlers();

        //配置Netty的serverBootstrap启动类
        //设置事件循环组
        serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                //设置是否使用Epoll模式
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                //设置服务端接收客户端连接队列的大小
                .option(ChannelOption.SO_BACKLOG, 1024)
                //设置服务端监听地址可重用标志
                .option(ChannelOption.SO_REUSEADDR, true)
                //设置启用TCP探活机制的标志
                .childOption(ChannelOption.SO_KEEPALIVE, false)
                //禁用Nagle算法，也就是说并不会因为要发送的数据包过小而合并数据包，从而延迟发送消息
                .childOption(ChannelOption.TCP_NODELAY, true)
                //设置服务端要监听的IP地址和端口号
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getBindAddress(),
                        this.nettyServerConfig.getListenPort()))
                //为接收到的客户端channel设置handler处理器，也就是为客户端channel设置pipeline
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        //配置接收到的客户端channel
                        configChannel(ch);
                    }
                });
        //为服务器设置自定义配置信息
        addCustomConfig(serverBootstrap);
        try {
            //在这里Netty服务器监听了对应的端口号
            ChannelFuture sync = serverBootstrap.bind().sync();
            //得到netty服务器监听的地址
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            //判断Netty配置信息对象中监听的端口号是否被更新了，如果没有被更新，那就把真正监听的端口号更新到配置信息对象中
            if (0 == nettyServerConfig.getListenPort()) {
                this.nettyServerConfig.setListenPort(addr.getPort());
            }
            //记录Netty服务器启动日志
            log.info("RemotingServer started, listening {}:{}", this.nettyServerConfig.getBindAddress(),
                    this.nettyServerConfig.getListenPort());
            //把当前启动的Netty服务器和服务器监听的端口号存储到remotingServerTable这个map中
            this.remotingServerTable.put(this.nettyServerConfig.getListenPort(), this);
        } catch (Exception e) {
            throw new IllegalStateException(String.format("Failed to bind to %s:%d", nettyServerConfig.getBindAddress(),
                    nettyServerConfig.getListenPort()), e);
        }
        //判断channelEventListener是否不为null，这个时候channelEventListener成员变量肯定不为null
        //创建当前类的对象时，这个channelEventListener成员变量就已经被NamesrvController中的brokerHousekeepingService成员变量赋值了
        if (this.channelEventListener != null) {
            //当监听器不为null的时候，就可以启动netty事件执行器了，这个执行器是当前类的父类中的成员变量
            //它其实就是一个单线程执行器，一旦被启动之后，就要不断从任务队列中获取任务，然后执行任务
            //那么要执行的是什么任务呢？任务队列中的任务又是什么时候被添加的呢？具体操作可以去当前类的NettyConnectManageHandler内部类中查看
            this.nettyEventExecutor.start();
        }

        //定义一个定时任务，这个定时任务会定期扫描响应表中的响应，用于扫描响应表中没有收到响应的ResponseFuture对象
        //其实就是判断哪些请求超时了，一直没有收到响应，如果有超时请求，那就从响应表中移除对应的ResponseFuture对象
        //不再关注该请求的结果
        TimerTask timerScanResponseTable = new TimerTask() {
            @Override
            public void run(Timeout timeout) {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                } finally {
                    //重新向时间轮中提交定时任务，每个1000毫秒执行一次
                    timer.newTimeout(this, 1000, TimeUnit.MILLISECONDS);
                }
            }
        };
        //把定时任务提交给时间轮，第一次提交的时候延迟3秒执行
        this.timer.newTimeout(timerScanResponseTable, 1000 * 3, TimeUnit.MILLISECONDS);

        //定时任务执行器定时打印Namesrv接收不同请求和回复响应的次数
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            try {
                NettyRemotingServer.this.printRemotingCodeDistribution();
            } catch (Throwable e) {
                TRAFFIC_LOGGER.error("NettyRemotingServer print remoting code distribution exception", e);
            }
        }, 1, 1, TimeUnit.SECONDS);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：配置服务端接收到的客户端channel的方法
     */
    protected ChannelPipeline configChannel(SocketChannel ch) {
        return ch.pipeline()
                //添加处理器与执行处理器中对应方法的默认执行器组
                .addLast(defaultEventExecutorGroup,
                        //设置编码器
                        encoder,
                        //设置解码器
                        new NettyDecoder(),
                        distributionHandler,
                        //设置空闲状态处理器， nettyServerConfig.getServerChannelMaxIdleTimeSeconds()就是从配置信息对象中得到channel最大的空闲时间
                        new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                        //设置客户端channel连接管理器
                        connectionManageHandler,
                        serverHandler);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：为Netty服务器设置一些用户自定义的配置信息
     */
    private void addCustomConfig(ServerBootstrap childHandler) {
        //设置服务器发送缓冲区的大小
        if (nettyServerConfig.getServerSocketSndBufSize() > 0) {
            log.info("server set SO_SNDBUF to {}", nettyServerConfig.getServerSocketSndBufSize());
            childHandler.childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize());
        }
        //设置服务器接收缓冲区的大小
        if (nettyServerConfig.getServerSocketRcvBufSize() > 0) {
            log.info("server set SO_RCVBUF to {}", nettyServerConfig.getServerSocketRcvBufSize());
            childHandler.childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize());
        }
        //设置写缓冲区的高水位线和低水位线
        if (nettyServerConfig.getWriteBufferLowWaterMark() > 0 && nettyServerConfig.getWriteBufferHighWaterMark() > 0) {
            log.info("server set netty WRITE_BUFFER_WATER_MARK to {},{}",
                    nettyServerConfig.getWriteBufferLowWaterMark(), nettyServerConfig.getWriteBufferHighWaterMark());
            childHandler.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                    nettyServerConfig.getWriteBufferLowWaterMark(), nettyServerConfig.getWriteBufferHighWaterMark()));
        }
        //判断是否开启了ByteBuf池化功能，如果开启了就设置到Netty中
        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：终止Namesrv服务器的方法
     */
    @Override
    public void shutdown() {
        try {
            //判断是否启用了优雅停机，如果启用了优雅停机，并且当前服务器还不是正在停机状态
            if (nettyServerConfig.isEnableShutdownGracefully() && isShuttingDown.compareAndSet(false, true)) {
                //那就等待一段时间再执行停机操作，这个时间也是从配置信息类中获取的，是可以由用户自己配置的
                Thread.sleep(Duration.ofSeconds(nettyServerConfig.getShutdownWaitTimeSeconds()).toMillis());
            }
            //停止时间轮
            this.timer.stop();
            //优雅关闭事件循环组
            this.eventLoopGroupBoss.shutdownGracefully();
            this.eventLoopGroupSelector.shutdownGracefully();
            //关闭netty事件执行器
            this.nettyEventExecutor.shutdown();
            //关闭处理默认事件的循环组
            if (this.defaultEventExecutorGroup != null) {
                this.defaultEventExecutorGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            log.error("NettyRemotingServer shutdown exception, ", e);
        }
        //关闭公共事件执行器
        if (this.publicExecutor != null) {
            try {
                this.publicExecutor.shutdown();
            } catch (Exception e) {
                log.error("NettyRemotingServer shutdown exception, ", e);
            }
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：注册请求处理器的方法，注册的时候会把对应的请求码，和处理请求的执行器都传递到该方法中，该方法会在第二版本代码中实现
     */
    @Override
    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        //判断传入的执行器是否为空
        ExecutorService executorThis = executor;
        if (null == executor) {
            //如果为空就是用公共执行器处理请求处理器接收到的请求
            executorThis = NettyRemotingServer.this.publicExecutor;
        }
        //这里创建了一个Pair对象，把请求处理器和请求处理器要使用的执行器封装到一起了
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
        //然后把请求码和请存放请求处理器的pair的映射存放到父类的processorTable中了
        this.processorTable.put(requestCode, pair);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：注册默认的请求处理器的方法，这个默认的请求处理器非常重要，在第二版本代码中，我也会为大家实现这个方法
     */
    @Override
    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executor) {
        this.defaultRequestProcessorPair = new Pair<>(processor, executor);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：为Namesrv服务端接收到的channel准备可共享的handler处理器的方法
     */
    private void prepareSharableHandlers() {
        encoder = new NettyEncoder();
        connectionManageHandler = new NettyConnectManageHandler();
        serverHandler = new NettyServerHandler();
        distributionHandler = new RemotingCodeDistributionHandler();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：打印Namesrv接收不同请求和回复响应次数的方法
     */
    private void printRemotingCodeDistribution() {
        if (distributionHandler != null) {
            //得到入站消息的字符串快照
            String inBoundSnapshotString = distributionHandler.getInBoundSnapshotString();
            //记录日志信息
            if (inBoundSnapshotString != null) {
                TRAFFIC_LOGGER.info("Port: {}, RequestCode Distribution: {}",
                        nettyServerConfig.getListenPort(), inBoundSnapshotString);
            }
            //得到出站消息的字符串快照
            String outBoundSnapshotString = distributionHandler.getOutBoundSnapshotString();
            //记录日志信息
            if (outBoundSnapshotString != null) {
                TRAFFIC_LOGGER.info("Port: {}, ResponseCode Distribution: {}",
                        nettyServerConfig.getListenPort(), outBoundSnapshotString);
            }
        }
    }



    //以下方法都是一些简单的get、set方法我就不为这些简单的方法添加注释了


    @Override
    public int localListenPort() {
        return this.nettyServerConfig.getListenPort();
    }


    @Override
    public Pair<NettyRequestProcessor, ExecutorService> getProcessorPair(int requestCode) {
        return this.processorTable.get(requestCode);
    }


    @Override
    public Pair<NettyRequestProcessor, ExecutorService> getDefaultProcessorPair() {
        return this.defaultRequestProcessorPair;
    }


    @Override
    public RemotingServer newRemotingServer(final int port) {
        return null;
    }


    @Override
    public void removeRemotingServer(final int port) {
        this.remotingServerTable.remove(port);
    }


    //同步发送请求的方法
    @Override
    public RemotingCommand invokeSync(final Channel channel, final RemotingCommand request, final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        return this.invokeSyncImpl(channel, request, timeoutMillis);
    }


    //异步发送请求的方法
    @Override
    public void invokeAsync(Channel channel, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        this.invokeAsyncImpl(channel, request, timeoutMillis, invokeCallback);
    }


    //单向发送请求的方法
    @Override
    public void invokeOneway(Channel channel, RemotingCommand request, long timeoutMillis) throws InterruptedException,
            RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        this.invokeOnewayImpl(channel, request, timeoutMillis);
    }


    @Override
    public ChannelEventListener getChannelEventListener() {
        return channelEventListener;
    }


    @Override
    public ExecutorService getCallbackExecutor() {
        return this.publicExecutor;
    }



    public DefaultEventExecutorGroup getDefaultEventExecutorGroup() {
        return defaultEventExecutorGroup;
    }


    public NettyEncoder getEncoder() {
        return encoder;
    }


    public NettyConnectManageHandler getConnectionManageHandler() {
        return connectionManageHandler;
    }


    public NettyServerHandler getServerHandler() {
        return serverHandler;
    }


    public RemotingCodeDistributionHandler getDistributionHandler() {
        return distributionHandler;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/27
     * @方法描述：这个内部类的对象就是channel的一个入站处理器，并且这个入站处理器就是专门用来处理Namesrv接收到的各种消息的
     */
    @ChannelHandler.Sharable
    public class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {


        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) {
            //得到Namesrv监听的本地端口号
            int localPort = RemotingHelper.parseSocketAddressPort(ctx.channel().localAddress());
            //根据端口号从remotingServerTable中得到当前接收入站消息的Namesrv服务器
            NettyRemotingAbstract remotingAbstract = NettyRemotingServer.this.remotingServerTable.get(localPort);
            if (localPort != -1 && remotingAbstract != null) {
                //如果找到了具体的服务器，那就直接把接收到的消息交给服务器处理
                remotingAbstract.processMessageReceived(ctx, msg);
                return;
            }
            //如果没有找到，那就意味着这个端口号对应的服务器可能停止工作了
            //直接关闭当前客户端channel即可
            RemotingHelper.closeChannel(ctx.channel());
        }

        //当channel的可写状态发生变更时，该方法就会被调用
        @Override
        public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
            if (channel.isWritable()) {
                //当通道可写的时候，就把自动读开关打开
                if (!channel.config().isAutoRead()) {
                    channel.config().setAutoRead(true);
                    log.info("Channel[{}] turns writable, bytes to buffer before changing channel to un-writable: {}",
                            RemotingHelper.parseChannelRemoteAddr(channel), channel.bytesBeforeUnwritable());
                }
            } else {//当通道不可写的时候，就把自动读开关关闭，所谓自动读就是自动读取channel中的数据
                //那为什么在通道不可写的时候，就把自动读功能禁用了呢？当channel不可写的时候，就意味着写缓冲区满了，如果这个时候再自动接收并处理更多的请求
                //那么请求处理完之后肯定要回复响应，也就意味着要把更多数据写入到写缓冲区中，但写缓冲区已经满了，channel不可写了，这个时候再自动读取并处理更多的请求
                //然后回复响应，无疑会增加写缓冲区的压力，这个时候应该先等一等，等写缓冲区可写了，压力小一些了，再打开自动读开关，处理接收到的请求
                //大家可以品味一下这个逻辑，体会这个操作的重要性
                channel.config().setAutoRead(false);
                log.warn("Channel[{}] auto-read is disabled, bytes to drain before it turns writable: {}",
                        RemotingHelper.parseChannelRemoteAddr(channel), channel.bytesBeforeWritable());
            }
            super.channelWritabilityChanged(ctx);
        }
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：这个类的对象就是客户端channel连接管理器，并且该对象也是一个Netty的handler，会被设置到接收到的客户端channel的pipeline中
     */
    @ChannelHandler.Sharable
    public class NettyConnectManageHandler extends ChannelDuplexHandler {

        //该方法会在接收到的客户端channel成功注册到一个单线程执行器的selector上之后被调用
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            //当一个客户端成功连接到当前服务器后，这行日志信息就会输出到控制台，我为大家提供的测试类中输出到控制台的就是这行信息
            log.info("NETTY SERVER PIPELINE: channelRegistered {}", remoteAddress);
            super.channelRegistered(ctx);
        }

        //该方法会在客户端channel取消注册的时候被调用，其实就是被关闭的时候
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelUnregistered, the channel[{}]", remoteAddress);
            super.channelUnregistered(ctx);
        }


        //该方法会在客户端channel和服务端连接断开时被回调
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            //得到客户端channel的远程地址
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            super.channelInactive(ctx);
            //这里执行了一个比较重要的操作，那就是判断channelEventListener监听器是否不为空，如果不为空，就意味着创建了
            //broker的channel管理器，也就是brokerHousekeepingService对象
            if (NettyRemotingServer.this.channelEventListener != null) {
                //一旦触发channel的关闭事件，当前方法就会被回调，也就意味着有客户端channel和服务端的连接断开了
                //这个时候就可以把断开的客户端channel封装到NettyEvent事件对象中，然后把这个事件存放到nettyEventExecutor执行器的任务队列中
                //nettyEventExecutor执行器是当前类的父类的一个成员变量
                //而nettyEventExecutor执行器处理任务队列中任务的时候，就会把任务取出来，交给channelEventListener监听器，也就是brokerHousekeepingService对象处理
                //因为是连接断开事件，所以broker的连接管理器就会把断开连接的channel从存储存活的broker的map中把对应的channel移除
                //这个逻辑现在大家也许不会理解，这没关系，大家只要先熟悉brokerHousekeepingService监听器对象的工作流程即可
                //具体的工作内容，等后面我会为大家全部实现
                NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
            }
        }

        //以下方法的逻辑同上，我就不再添加详细注释了，大家自己看看就行
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //得到客户端channel的远程地址
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            //把channel信息输出到控制台，我为大家提供的测试类中输出到控制台的也包含这行信息
            log.info("NETTY SERVER PIPELINE: channelActive, the channel[{}]", remoteAddress);
            super.channelActive(ctx);
            if (NettyRemotingServer.this.channelEventListener != null) {

                NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remoteAddress, ctx.channel()));
            }
        }



        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {

            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    log.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
                    RemotingHelper.closeChannel(ctx.channel());
                    if (NettyRemotingServer.this.channelEventListener != null) {
                        NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
                    }
                }
            }
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.warn("NETTY SERVER PIPELINE: exceptionCaught {}", remoteAddress);
            log.warn("NETTY SERVER PIPELINE: exceptionCaught exception.", cause);
            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
            }
            RemotingHelper.closeChannel(ctx.channel());
        }
    }

}
